package olap4cloud.hadop.statistic;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import olap4cloud.data.DataStruct;
import olap4cloud.util2.MyDateUtil;
import olap4coud.report.DataReport;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Increment;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.MD5Hash;

public class StatisticsTable {

	public static final String tableName = "statistics";
	public static final byte[] columfHour = Bytes.toBytes("h");
	public static final byte[] columfDate = Bytes.toBytes("d");

	public static byte[] hashKey(long time) {
		return Bytes.toBytes(MD5Hash.getMD5AsHex(Bytes.toBytes(time)));
	}

	public static byte[] hashKey(byte[] b) {
		return Bytes.toBytes(MD5Hash.getMD5AsHex(b));
	}

	public static Increment increDataInHbaseByHour(DataStruct dataStruct) {
		long hourlong = MyDateUtil.roundDownHourTime(dataStruct.getTimestamp());
		Increment incre = new Increment(hashKey(hourlong));
		incre.addColumn(columfHour,
				Bytes.toBytes("d_" + dataStruct.getDomain()), 1l);
		incre.addColumn(columfHour,
				Bytes.toBytes("c_" + dataStruct.getChannelId()), 1l);
		incre.addColumn(columfHour,
				Bytes.toBytes("b_" + dataStruct.getBrowser()), 1l);
		incre.addColumn(columfHour, Bytes.toBytes("total"), 1l);
		return incre;
	}

	public static Increment increDataByDomainInHbaseByHour(DataStruct dataStruct) {
		long hourlong = MyDateUtil.roundDownHourTime(dataStruct.getTimestamp());
		Increment incre = new Increment((Bytes.add(Bytes.toBytes(hourlong),
				Bytes.toBytes(dataStruct.getDomain()))));
		incre.addColumn(columfHour,
				Bytes.toBytes("c_" + dataStruct.getChannelId()), 1l);
		incre.addColumn(columfHour,
				Bytes.toBytes("b_" + dataStruct.getBrowser()), 1l);
		incre.addColumn(columfHour, Bytes.toBytes("total"), 1l);
		return incre;
	}

	public static Increment increDataInHbaseByDate(DataStruct dataStruct) {
		long datelong = MyDateUtil.roundDownDateTime(dataStruct.getTimestamp());
		Increment incre = new Increment(hashKey(datelong));
		incre.addColumn(columfDate,
				Bytes.toBytes("d_" + dataStruct.getDomain()), 1l);
		incre.addColumn(columfDate,
				Bytes.toBytes("c_" + dataStruct.getChannelId()), 1l);
		incre.addColumn(columfDate,
				Bytes.toBytes("b_" + dataStruct.getBrowser()), 1l);
		incre.addColumn(columfDate, Bytes.toBytes("total"), 1l);
		return incre;
	}

	public static String getTableName() {
		return tableName;
	}

	/**
	 * @uml.property  name="htable"
	 * @uml.associationEnd  
	 */
	HTable htable;
	/**
	 * @uml.property  name="conf"
	 * @uml.associationEnd  multiplicity="(1 1)"
	 */
	Configuration conf;

	public StatisticsTable() {
		// TODO Auto-generated constructor stub
		conf = HBaseConfiguration.create(new Configuration());
	}

	public void tryConnect() {
		try {
			htable = new HTable(conf, StatisticsTable.getTableName());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public List<DataReport> reportDomainByHour(long hourLong) {
		tryConnect();
		List<DataReport> result = new ArrayList<DataReport>();
		Get get = new Get(hashKey(hourLong));
		get.addFamily(columfHour);
		try {
			Result rs = htable.get(get);
			for (KeyValue kv : rs.raw()) {
				DataReport dataReport = getReportHbase(kv, "d");
				if (dataReport != null)
					result.add(dataReport);
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return result;

	}

	public List<DataReport> reportDomainByDate(long dateLong) {
		tryConnect();
		List<DataReport> result = new ArrayList<DataReport>();
		Get get = new Get(Bytes.toBytes(dateLong+""));
		get.addFamily(columfDate);
		try {
			Result rs = htable.get(get);
			for (KeyValue kv : rs.raw()) {
				DataReport dataReport = getReportHbase(kv, "d");
				if (dataReport != null)
					result.add(dataReport);
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return result;

	}

	public List<DataReport> reportBrowserByHour(long hourLong) {
		tryConnect();
		List<DataReport> result = new ArrayList<DataReport>();
		Get get = new Get(hashKey(hourLong));
		get.addFamily(columfHour);
		try {
			Result rs = htable.get(get);
			for (KeyValue kv : rs.raw()) {
				DataReport dataReport = getReportHbase(kv, "b");
				if (dataReport != null)
					result.add(dataReport);
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return result;

	}

	public List<DataReport> reportBrowserByDate(long dateLong) {
		tryConnect();
		List<DataReport> result = new ArrayList<DataReport>();
		Get get = new Get(hashKey(dateLong));
		get.addFamily(columfDate);
		try {
			Result rs = htable.get(get);
			for (KeyValue kv : rs.raw()) {
				DataReport dataReport = getReportHbase(kv, "b");
				if (dataReport != null)
					result.add(dataReport);
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return result;

	}

	public List<DataReport> reportChannelByHour(long hourLong) {
		tryConnect();
		List<DataReport> result = new ArrayList<DataReport>();
		Get get = new Get(hashKey(hourLong));
		get.addFamily(columfHour);
		try {
			Result rs = htable.get(get);
			for (KeyValue kv : rs.raw()) {
				DataReport dataReport = getReportHbase(kv, "c");
				if (dataReport != null)
					result.add(dataReport);
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return result;

	}

	public List<DataReport> reportChannelByDate(long dateLong) {
		tryConnect();
		List<DataReport> result = new ArrayList<DataReport>();
		Get get = new Get(hashKey(dateLong));
		get.addFamily(columfDate);
		try {
			Result rs = htable.get(get);
			for (KeyValue kv : rs.raw()) {
				DataReport dataReport = getReportHbase(kv, "c");
				if (dataReport != null)
					result.add(dataReport);
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return result;

	}

	public DataReport getReportHbase(KeyValue kv, String pre) {
		String value = Bytes.toString(kv.getQualifier());
		if (value.contains(pre)) {
			long pageview = Bytes.toLong(kv.getValue());
			return new DataReport(value, pageview);
		}
		return null;
	}

}
